package com.yanyeori.framework.redis.disruptor.producer;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.yanyeori.framework.redis.config.ConfigFactory;
import com.yanyeori.framework.redis.disruptor.event.RedisPushEvent;
import com.yanyeori.framework.redis.disruptor.handle.RedisPushEventHandler;
import com.yanyeori.framework.redis.model.PushMessage;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Executors;

/**
 * redis事件生产者
 *
 * @author chenkuan 2022/5/9
 */
@Slf4j
public class RedisPushEventProducer {

    private static volatile RedisPushEventProducer instance;
    private RingBuffer<RedisPushEvent> ringBuffer;

    private RedisPushEventProducer() {
        init();
    }

    public static RedisPushEventProducer getInstance() {
        if (instance == null) {
            synchronized (RedisPushEventProducer.class) {
                if (instance == null) {
                    instance = new RedisPushEventProducer();
                }
            }
        }
        return instance;
    }

    public void send(PushMessage message) {
        //获取下一个Event槽的下标
        long sequence = ringBuffer.next();
        try {
            //sequence位置取出的事件是空事件
            RedisPushEvent event = ringBuffer.get(sequence);
            //给Event填充数据
            event.setTopic(message.getTopic());
            event.setData(message.getData());
            log.debug("Send message to redisEventRingBuffer. sequence: {}", sequence);
        } catch (Exception e) {
            log.error("Failed to Send message to redisEventRingBuffer. sequence: {}", sequence, e);
        } finally {
            //注意最后的publish方法必须放在finally中以确保必须得到调用；如果某个请求的sequence未被提交将会堵塞后续的发布操作或者其他的producer
            ringBuffer.publish(sequence);
        }
    }

    private void init() {
        //初始化disruptor
        Disruptor<RedisPushEvent> disruptor = new Disruptor<>(RedisPushEvent::new, 1024 * 1024, Executors.defaultThreadFactory());
        //设置事件业务处理器---消费者
        disruptor.handleEventsWith(getHandler(ConfigFactory.getConfig().getEvent().getHandlerSize()));
        //启动disruptor线程
        disruptor.start();
        //获取ringBuffer环，用于接取生产者生产的事件
        ringBuffer = disruptor.getRingBuffer();
    }

    private EventHandler<? super RedisPushEvent>[] getHandler(int size) {
        RedisPushEventHandler[] handlers = new RedisPushEventHandler[size];
        for (int i = 0; i < size; i++) {
            handlers[i] = new RedisPushEventHandler(i);
        }
        return handlers;
    }
}
